package com.roncoo.eshop.storm.zk;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Copyright: Copyright (c) 2019 Hu-Cheng
 *
 * @ClassName: ZookeeperSession.java
 * @Description: zookeeper工具类
 *
 * @version: v1.0.0
 * @author: hucheng
 * @date: Oct 16, 2019 11:19:40 AM
 *
 */
public class ZookeeperSession {

	private ZooKeeper zooKeeper;

	private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperSession.class);

	private CountDownLatch countDownLatch = new CountDownLatch(1);

	/**
	 * 构造方法初始化zookeeper链接
	 */
	public ZookeeperSession() {
		try {
			this.zooKeeper = new ZooKeeper("192.168.124.204:2181,192.16.124.205:2181,192.168.124.206:2181", 50000,
					new ZookeeperWatcher());
			try {
				// CountDownLatch
				// java多线程并发同步的一个工具类
				// 会传递一些数据，比如123都可以
				// 然后await(),如果数字不是0，那么就会卡住，等等
				// 其他线程可以拥有countDown(),减1
				// 如果数字减到0，那么之前所有在await()的进程，都会逃出阻塞状态
				// 继续向下执行
				countDownLatch.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 根据用户传入的path，获取分布式锁
	 * 
	 * @param path
	 */
	public void acquireDistributedLockByPath(String path) {
		try {
			zooKeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
			LOGGER.info("success to acquire lock for " + path);
		} catch (Exception e) {
			int count = 0;
			while (true) {
				try {
					Thread.sleep(1000);
					zooKeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
				} catch (Exception e1) {
					count++;
					LOGGER.info("the " + count + " times try to acquire lock for " + path + "......");
					continue;
				}
				LOGGER.info("success to acquire lock for after " + count + " times try......");
			}
		}
	}

	/**
	 * 根据path释放掉zookeeper锁
	 * 
	 * @param path
	 */
	public void releaseDistributedLock(String path) {
		try {
			zooKeeper.delete(path, -1);
			LOGGER.info("release the lock for " + path);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 获取zookeeper节点数据
	 * 
	 * @param path
	 * @return
	 */
	public String getNodeData(String path) {
		try {
			return new String(zooKeeper.getData(path, false, new Stat()));
		} catch (Exception e) {
			e.printStackTrace();
		}
		return "";
	}

	/**
	 * 创建zookeeper节点
	 * 
	 * @param path
	 */
	public void createNode(String path) {
		try {
			Stat stat = zooKeeper.exists(path, true);
			if (stat == null) {
				zooKeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 往zookeeper节点中存入数据
	 * 
	 * @param path
	 * @param data
	 */
	public void setNodeData(String path, String data) {
		try {
			zooKeeper.setData(path, data.getBytes(), -1);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * Copyright: Copyright (c) 2019 Hu-Cheng
	 *
	 * @ClassName: ZookeeperSession.java
	 * @Description: 建立zk，session的watcher
	 *
	 * @version: v1.0.0
	 * @author: hucheng
	 * @date: Oct 16, 2019 11:33:44 AM
	 *
	 */
	private class ZookeeperWatcher implements Watcher {
		@Override
		public void process(WatchedEvent event) {
			LOGGER.info("Receive watcher event : " + event.getState());
			if (KeeperState.SyncConnected == event.getState()) {
				countDownLatch.countDown();
			}
		}
	}

	private static class Singleton {

		private static ZookeeperSession zookeeperSession;

		static {
			zookeeperSession = new ZookeeperSession();
		}

		public static ZookeeperSession getInstance() {
			return zookeeperSession;
		}

	}

	public static ZookeeperSession getInstance() {
		return Singleton.getInstance();
	}

}
